package com.mtiiot.mq.service;



import com.mtiiot.mq.esdao.BaseElasticsearch;
import com.mtiiot.mq.utils.Contants;
import net.sf.json.JSONObject;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.codehaus.jackson.map.ObjectMapper;

import java.util.List;

/**
 * Created by Administrator on 16-6-22.
 */
//@Component("logAlerted")
public class RedisMqConsumerTask {
    private static final Logger logger = LogManager.getLogger(RedisMqConsumerTask.class);
    private LogService1 logService1;
    private RedisService1 redisService1;
    private BaseElasticsearch baseElasticsearch;

    public BaseElasticsearch getBaseElasticsearch() {
        return baseElasticsearch;
    }

    public void setBaseElasticsearch(BaseElasticsearch baseElasticsearch) {
        this.baseElasticsearch = baseElasticsearch;
    }

    public LogService1 getLogService1() {
        return logService1;
    }

    public void setLogService1(LogService1 logService1) {
        this.logService1 = logService1;
    }



    public synchronized void  execute() {
        //logger.info("========redismq定时任务开始执行==========");
        try {
            Long len = redisService1.llen(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "RedisMqConsumerTask", "execute"),Contants.METADATA_REDISKEY_001);
            if (len > 0) {
                List<String> list = redisService1.lrange(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "RedisMqConsumerTask", "execute"),Contants.METADATA_REDISKEY_001, 0, len);
                for (String str : list) {
                    Long result=redisService1.lrem(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "RedisMqConsumerTask", "execute"),Contants.METADATA_REDISKEY_001,-1,str);
                    if(result!=1){
                        //删除失败
                        logger.warn("================redis 删除数据失败!!=========");
                        continue;
                    }
                    MqInfo mqInfo=null;
                    ObjectMapper mapper = new ObjectMapper();
                    try {
                            mqInfo = mapper.readValue(str, MqInfo.class);
                            if(Contants.TOPIC_METADATAINFO.equals(mqInfo.getTopic())){
                                // 元数据处理
                                String type=mqInfo.getKey();
                                JSONObject bean = JSONObject.fromObject(mqInfo.getValue());
                                String id = bean.getString("id");
                                String json=mapper.writeValueAsString(mqInfo.getValue());
                                if(StringUtil.isEmpty(type)){
                                    throw new Exception("消息队列的key不能为空!!");
                                }
                                if(bean==null){
                                    throw new Exception("消息队列的value不能为空!!");
                                }
                                if(StringUtil.isEmpty(id)){
                                    throw new Exception("消息队列的value转bean对应的id属性为空!");
                                }
                                baseElasticsearch.createIndexById(Contants.INDEX_METADATA,type,json,id);
                            } else {
                                // 业务数据处理
                                String type=mqInfo.getKey();
                                String[] arr = type.split("%");
                                String seqNo = arr[1];
                                String topic = mqInfo.getTopic();
                                String body = mapper.writeValueAsString(mqInfo.getValue());
                                // 调用入库的包
                                BaseService.sendData(topic, seqNo, body);
                            }
                    } catch (Exception ex) {
                        if(mqInfo.getReTryTime()<3){
                            mqInfo.setReTryTime(mqInfo.getReTryTime()+1);
                            redisService1.lpush(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "RedisMqConsumerTask", "execute"),Contants.METADATA_REDISKEY_001,mapper.writeValueAsString(mqInfo));
                        }else{
                            redisService1.lpush(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "RedisMqConsumerTask", "execute"),Contants.METADATA_REDISKEY_001_ERROR, mapper.writeValueAsString(mqInfo));
                            logService1.sendLog(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "RedisMqConsumerTask", "execute"),LogLevel.ERROR," after reTry 3 Times,also failure!!",ex.getMessage());
                            logger.error(ex.getMessage(), ex);
                        }
                    }
                }
            }else{
                //logger.info("本次redismq定时任务没有要执行的任务");
            }
        } catch (Exception e) {
            logService1.sendLog(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "RedisMqConsumerTask", "execute"),LogLevel.ERROR,"",e.getMessage());
            logger.error(e.getMessage(), e);
        }
    }

    public RedisService1 getRedisService1() {
        return redisService1;
    }

    public void setRedisService1(RedisService1 redisService1) {
        this.redisService1 = redisService1;
    }

}
